Repository: dragonflyoss/Dragonfly2
Pull request #544 (2021-08-12 14:05:09)

feat: rename task to job
Signed-off-by: Gaius <gaius.qi@gmail.com>

<!--- Provide a general summary of your changes in the Title above -->

## Description
- Rename task to job.
<!--- 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! -->

- [x] Code compiles correctly.
- [x] My code follows the code style of this project.
- [ ] My change requires a change to the documentation.
- [ ] I have updated the documentation accordingly.
- [x] I have read the **CONTRIBUTING** document.
- [x] I have added tests to cover my changes.
- [x] All new and existing tests passed.



Patch:
diff --git a/client/daemon/peer/peertask_base.go b/client/daemon/peer/peertask_base.go
index 03f37a0a..98b875f1 100644
--- a/client/daemon/peer/peertask_base.go
+++ b/client/daemon/peer/peertask_base.go
@@ -24,10 +24,9 @@ import (
 	"sync"
 	"time"
 
-	"go.uber.org/atomic"
-
 	"github.com/pkg/errors"
 	"go.opentelemetry.io/otel/trace"
+	"go.uber.org/atomic"
 	"golang.org/x/time/rate"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -55,6 +54,8 @@ const (
 
 var errPeerPacketChanged = errors.New("peer packet changed")
 
+var _ Task = (*peerTask)(nil)
+
 type peerTask struct {
 	*logger.SugaredLoggerOnWith
 	ctx    context.Context
@@ -64,7 +65,7 @@ type peerTask struct {
 	needBackSource bool
 
 	backSourceFunc        func()
-	reportPieceResultFunc func(pieceTask *base.PieceInfo, pieceResult *scheduler.PieceResult) error
+	reportPieceResultFunc func(result *pieceTaskResult) error
 	setContentLengthFunc  func(i int64) error
 
 	request *scheduler.PeerTaskRequest
@@ -128,10 +129,14 @@ type peerTask struct {
 	limiter *rate.Limiter
 }
 
-var _ Task = (*peerTask)(nil)
+type pieceTaskResult struct {
+	piece       *base.PieceInfo
+	pieceResult *scheduler.PieceResult
+	err         error
+}
 
-func (pt *peerTask) ReportPieceResult(pieceTask *base.PieceInfo, pieceResult *scheduler.PieceResult) error {
-	return pt.reportPieceResultFunc(pieceTask, pieceResult)
+func (pt *peerTask) ReportPieceResult(result *pieceTaskResult) error {
+	return pt.reportPieceResultFunc(result)
 }
 
 func (pt *peerTask) SetCallback(callback TaskCallback) {
@@ -446,6 +451,7 @@ loop:
 		limit = 1
 		// get failed piece
 		if num, ok = pt.waitFailedPiece(); !ok {
+			// when ok == false, indicates than need break loop
 			break loop
 		}
 	}
@@ -592,7 +598,7 @@ func (pt *peerTask) waitFailedPiece() (int32, bool) {
 		return -1, false
 	case failed := <-pt.failedPieceCh:
 		pt.Warnf("download piece/%d failed, retry", failed)
-		return -1, false
+		return -1, true
 	}
 }
 
@@ -616,8 +622,9 @@ func (pt *peerTask) downloadPieceWorker(id int32, pti Task, requests chan *Downl
 					pt.Errorf("request limiter error: %s", err)
 					waitSpan.RecordError(err)
 					waitSpan.End()
-					pti.ReportPieceResult(request.piece,
-						&scheduler.PieceResult{
+					pti.ReportPieceResult(&pieceTaskResult{
+						piece: request.piece,
+						pieceResult: &scheduler.PieceResult{
 							TaskId:        pt.GetTaskID(),
 							SrcPid:        pt.GetPeerID(),
 							DstPid:        request.DstPid,
@@ -626,7 +633,9 @@ func (pt *peerTask) downloadPieceWorker(id int32, pti Task, requests chan *Downl
 							Code:          dfcodes.ClientRequestLimitFail,
 							HostLoad:      nil,
 							FinishedCount: 0, // update by peer task
-						})
+						},
+						err: err,
+					})
 					pt.failedReason = err.Error()
 					pt.failedCode = dfcodes.ClientRequestLimitFail
 					pt.cancel()
diff --git a/client/daemon/peer/peertask_dummy.go b/client/daemon/peer/peertask_dummy.go
index eccb20bd..9504fa2f 100644
--- a/client/daemon/peer/peertask_dummy.go
+++ b/client/daemon/peer/peertask_dummy.go
@@ -19,11 +19,12 @@ package peer
 import (
 	"context"
 
+	"google.golang.org/grpc"
+
 	"d7y.io/dragonfly/v2/internal/dfcodes"
 	"d7y.io/dragonfly/v2/internal/dferrors"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	schedulerclient "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client"
-	"google.golang.org/grpc"
 )
 
 type dummySchedulerClient struct {
diff --git a/client/daemon/peer/peertask_file.go b/client/daemon/peer/peertask_file.go
index badb4dcf..c42daedb 100644
--- a/client/daemon/peer/peertask_file.go
+++ b/client/daemon/peer/peertask_file.go
@@ -216,38 +216,38 @@ func (pt *filePeerTask) Start(ctx context.Context) (chan *FilePeerTaskProgress,
 	return pt.progressCh, nil
 }
 
-func (pt *filePeerTask) ReportPieceResult(piece *base.PieceInfo, pieceResult *scheduler.PieceResult) error {
+func (pt *filePeerTask) ReportPieceResult(result *pieceTaskResult) error {
 	// goroutine safe for channel and send on closed channel
 	defer pt.recoverFromPanic()
-	pt.Debugf("report piece %d result, success: %t", piece.PieceNum, pieceResult.Success)
+	pt.Debugf("report piece %d result, success: %t", result.piece.PieceNum, result.pieceResult.Success)
 
 	// retry failed piece
-	if !pieceResult.Success {
-		pieceResult.FinishedCount = pt.readyPieces.Settled()
-		_ = pt.peerPacketStream.Send(pieceResult)
-		pt.failedPieceCh <- pieceResult.PieceNum
-		pt.Errorf("%d download failed, retry later", piece.PieceNum)
+	if !result.pieceResult.Success {
+		result.pieceResult.FinishedCount = pt.readyPieces.Settled()
+		_ = pt.peerPacketStream.Send(result.pieceResult)
+		pt.failedPieceCh <- result.pieceResult.PieceNum
+		pt.Errorf("%d download failed, retry later", result.piece.PieceNum)
 		return nil
 	}
 
 	pt.lock.Lock()
-	if pt.readyPieces.IsSet(pieceResult.PieceNum) {
+	if pt.readyPieces.IsSet(result.pieceResult.PieceNum) {
 		pt.lock.Unlock()
-		pt.Warnf("piece %d is already reported, skipped", pieceResult.PieceNum)
+		pt.Warnf("piece %d is already reported, skipped", result.pieceResult.PieceNum)
 		return nil
 	}
 	// mark piece processed
-	pt.readyPieces.Set(pieceResult.PieceNum)
-	pt.completedLength.Add(int64(piece.RangeSize))
+	pt.readyPieces.Set(result.pieceResult.PieceNum)
+	pt.completedLength.Add(int64(result.piece.RangeSize))
 	pt.lock.Unlock()
 
-	pieceResult.FinishedCount = pt.readyPieces.Settled()
-	_ = pt.peerPacketStream.Send(pieceResult)
+	result.pieceResult.FinishedCount = pt.readyPieces.Settled()
+	_ = pt.peerPacketStream.Send(result.pieceResult)
 	// send progress first to avoid close channel panic
 	p := &FilePeerTaskProgress{
 		State: &ProgressState{
-			Success: pieceResult.Success,
-			Code:    pieceResult.Code,
+			Success: result.pieceResult.Success,
+			Code:    result.pieceResult.Code,
 			Msg:     "downloading",
 		},
 		TaskID:          pt.taskID,
diff --git a/client/daemon/peer/peertask_manager.go b/client/daemon/peer/peertask_manager.go
index f10a2cbd..3fb75d5e 100644
--- a/client/daemon/peer/peertask_manager.go
+++ b/client/daemon/peer/peertask_manager.go
@@ -62,7 +62,7 @@ type TaskManager interface {
 type Task interface {
 	Context() context.Context
 	Log() *logger.SugaredLoggerOnWith
-	ReportPieceResult(pieceTask *base.PieceInfo, pieceResult *scheduler.PieceResult) error
+	ReportPieceResult(result *pieceTaskResult) error
 	GetPeerID() string
 	GetTaskID() string
 	GetTotalPieces() int32
diff --git a/client/daemon/peer/peertask_stream.go b/client/daemon/peer/peertask_stream.go
index b730a292..3cd9d6ec 100644
--- a/client/daemon/peer/peertask_stream.go
+++ b/client/daemon/peer/peertask_stream.go
@@ -181,30 +181,30 @@ func newStreamPeerTask(ctx context.Context,
 	return ctx, pt, nil, nil
 }
 
-func (s *streamPeerTask) ReportPieceResult(piece *base.PieceInfo, pieceResult *scheduler.PieceResult) error {
+func (s *streamPeerTask) ReportPieceResult(result *pieceTaskResult) error {
 	defer s.recoverFromPanic()
 	// retry failed piece
-	if !pieceResult.Success {
-		_ = s.peerPacketStream.Send(pieceResult)
-		s.failedPieceCh <- pieceResult.PieceNum
+	if !result.pieceResult.Success {
+		_ = s.peerPacketStream.Send(result.pieceResult)
+		s.failedPieceCh <- result.pieceResult.PieceNum
 		return nil
 	}
 
 	s.lock.Lock()
-	if s.readyPieces.IsSet(pieceResult.PieceNum) {
+	if s.readyPieces.IsSet(result.pieceResult.PieceNum) {
 		s.lock.Unlock()
-		s.Warnf("piece %d is already reported, skipped", pieceResult.PieceNum)
+		s.Warnf("piece %d is already reported, skipped", result.pieceResult.PieceNum)
 		return nil
 	}
 	// mark piece processed
-	s.readyPieces.Set(pieceResult.PieceNum)
-	s.completedLength.Add(int64(piece.RangeSize))
+	s.readyPieces.Set(result.pieceResult.PieceNum)
+	s.completedLength.Add(int64(result.piece.RangeSize))
 	s.lock.Unlock()
 
-	pieceResult.FinishedCount = s.readyPieces.Settled()
-	_ = s.peerPacketStream.Send(pieceResult)
-	s.successPieceCh <- piece.PieceNum
-	s.Debugf("success piece %d sent", piece.PieceNum)
+	result.pieceResult.FinishedCount = s.readyPieces.Settled()
+	_ = s.peerPacketStream.Send(result.pieceResult)
+	s.successPieceCh <- result.piece.PieceNum
+	s.Debugf("success piece %d sent", result.piece.PieceNum)
 	select {
 	case <-s.ctx.Done():
 		s.Warnf("peer task context done due to %s", s.ctx.Err())
@@ -367,7 +367,7 @@ func (s *streamPeerTask) finish() error {
 		// send EOF piece result to scheduler
 		_ = s.peerPacketStream.Send(
 			scheduler.NewEndPieceResult(s.taskID, s.peerID, s.readyPieces.Settled()))
-		s.Debugf("end piece result sent")
+		s.Debugf("end piece result sent, peer task finished")
 		close(s.done)
 		//close(s.successPieceCh)
 		if err := s.callback.Done(s); err != nil {
@@ -385,7 +385,7 @@ func (s *streamPeerTask) cleanUnfinished() {
 		// send EOF piece result to scheduler
 		_ = s.peerPacketStream.Send(
 			scheduler.NewEndPieceResult(s.taskID, s.peerID, s.readyPieces.Settled()))
-		s.Debugf("end piece result sent")
+		s.Errorf("end piece result sent, peer task failed")
 		close(s.done)
 		//close(s.successPieceCh)
 		if err := s.callback.Fail(s, s.failedCode, s.failedReason); err != nil {
diff --git a/client/daemon/peer/piece_manager.go b/client/daemon/peer/piece_manager.go
index fbdbbca9..0b8dc6f6 100644
--- a/client/daemon/peer/piece_manager.go
+++ b/client/daemon/peer/piece_manager.go
@@ -94,6 +94,7 @@ func (pm *pieceManager) DownloadPiece(ctx context.Context, pt Task, request *Dow
 	var (
 		start = time.Now().UnixNano()
 		end   int64
+		err   error
 	)
 	defer func() {
 		_, rspan := tracer.Start(ctx, config.SpanPushPieceResult)
@@ -101,14 +102,14 @@ func (pm *pieceManager) DownloadPiece(ctx context.Context, pt Task, request *Dow
 		if success {
 			pm.pushSuccessResult(pt, request.DstPid, request.piece, start, end)
 		} else {
-			pm.pushFailResult(pt, request.DstPid, request.piece, start, end)
+			pm.pushFailResult(pt, request.DstPid, request.piece, start, end, err)
 		}
 		rspan.End()
 	}()
 
 	// 1. download piece from other peers
 	if pm.Limiter != nil {
-		if err := pm.Limiter.WaitN(ctx, int(request.piece.RangeSize)); err != nil {
+		if err = pm.Limiter.WaitN(ctx, int(request.piece.RangeSize)); err != nil {
 			pt.Log().Errorf("require rate limit access error: %s", err)
 			return
 		}
@@ -118,7 +119,12 @@ func (pm *pieceManager) DownloadPiece(ctx context.Context, pt Task, request *Dow
 	span.SetAttributes(config.AttributeTargetPeerID.String(request.DstPid))
 	span.SetAttributes(config.AttributeTargetPeerAddr.String(request.DstAddr))
 	span.SetAttributes(config.AttributePiece.Int(int(request.piece.PieceNum)))
-	r, c, err := pm.pieceDownloader.DownloadPiece(ctx, request)
+
+	var (
+		r io.Reader
+		c io.Closer
+	)
+	r, c, err = pm.pieceDownloader.DownloadPiece(ctx, request)
 	if err != nil {
 		span.RecordError(err)
 		span.End()
@@ -129,7 +135,8 @@ func (pm *pieceManager) DownloadPiece(ctx context.Context, pt Task, request *Dow
 	defer c.Close()
 
 	// 2. save to storage
-	n, err := pm.storageManager.WritePiece(ctx, &storage.WritePieceRequest{
+	var n int64
+	n, err = pm.storageManager.WritePiece(ctx, &storage.WritePieceRequest{
 		PeerTaskMetaData: storage.PeerTaskMetaData{
 			PeerID: pt.GetPeerID(),
 			TaskID: pt.GetTaskID(),
@@ -160,38 +167,44 @@ func (pm *pieceManager) DownloadPiece(ctx context.Context, pt Task, request *Dow
 
 func (pm *pieceManager) pushSuccessResult(peerTask Task, dstPid string, piece *base.PieceInfo, start int64, end int64) {
 	err := peerTask.ReportPieceResult(
-		piece,
-		&scheduler.PieceResult{
-			TaskId:        peerTask.GetTaskID(),
-			SrcPid:        peerTask.GetPeerID(),
-			DstPid:        dstPid,
-			PieceNum:      piece.PieceNum,
-			BeginTime:     uint64(start),
-			EndTime:       uint64(end),
-			Success:       true,
-			Code:          dfcodes.Success,
-			HostLoad:      nil, // TODO(jim): update host load
-			FinishedCount: 0,   // update by peer task
+		&pieceTaskResult{
+			piece: piece,
+			pieceResult: &scheduler.PieceResult{
+				TaskId:        peerTask.GetTaskID(),
+				SrcPid:        peerTask.GetPeerID(),
+				DstPid:        dstPid,
+				PieceNum:      piece.PieceNum,
+				BeginTime:     uint64(start),
+				EndTime:       uint64(end),
+				Success:       true,
+				Code:          dfcodes.Success,
+				HostLoad:      nil, // TODO(jim): update host load
+				FinishedCount: 0,   // update by peer task
+			},
+			err: nil,
 		})
 	if err != nil {
 		peerTask.Log().Errorf("report piece task error: %v", err)
 	}
 }
 
-func (pm *pieceManager) pushFailResult(peerTask Task, dstPid string, piece *base.PieceInfo, start int64, end int64) {
-	err := peerTask.ReportPieceResult(
-		piece,
-		&scheduler.PieceResult{
-			TaskId:        peerTask.GetTaskID(),
-			SrcPid:        peerTask.GetPeerID(),
-			DstPid:        dstPid,
-			PieceNum:      piece.PieceNum,
-			BeginTime:     uint64(start),
-			EndTime:       uint64(end),
-			Success:       false,
-			Code:          dfcodes.ClientPieceDownloadFail,
-			HostLoad:      nil,
-			FinishedCount: 0, // update by peer task
+func (pm *pieceManager) pushFailResult(peerTask Task, dstPid string, piece *base.PieceInfo, start int64, end int64, err error) {
+	err = peerTask.ReportPieceResult(
+		&pieceTaskResult{
+			piece: piece,
+			pieceResult: &scheduler.PieceResult{
+				TaskId:        peerTask.GetTaskID(),
+				SrcPid:        peerTask.GetPeerID(),
+				DstPid:        dstPid,
+				PieceNum:      piece.PieceNum,
+				BeginTime:     uint64(start),
+				EndTime:       uint64(end),
+				Success:       false,
+				Code:          dfcodes.ClientPieceDownloadFail,
+				HostLoad:      nil,
+				FinishedCount: 0, // update by peer task
+			},
+			err: err,
 		})
 	if err != nil {
 		peerTask.Log().Errorf("report piece task error: %v", err)
@@ -208,6 +221,7 @@ func (pm *pieceManager) processPieceFromSource(pt Task,
 		success bool
 		start   = time.Now().UnixNano()
 		end     int64
+		err     error
 	)
 
 	var (
@@ -236,12 +250,12 @@ func (pm *pieceManager) processPieceFromSource(pt Task,
 					PieceMd5:    "",
 					PieceOffset: pieceOffset,
 					PieceStyle:  0,
-				}, start, end)
+				}, start, end, err)
 		}
 	}()
 
 	if pm.Limiter != nil {
-		if err := pm.Limiter.WaitN(pt.Context(), int(size)); err != nil {
+		if err = pm.Limiter.WaitN(pt.Context(), int(size)); err != nil {
 			pt.Log().Errorf("require rate limit access error: %s", err)
 			return 0, err
 		}
@@ -249,7 +263,8 @@ func (pm *pieceManager) processPieceFromSource(pt Task,
 	if pm.calculateDigest {
 		reader = digestutils.NewDigestReader(reader)
 	}
-	n, err := pm.storageManager.WritePiece(
+	var n int64
+	n, err = pm.storageManager.WritePiece(
 		pt.Context(),
 		&storage.WritePieceRequest{
 			UnknownLength: unknownLength,
diff --git a/deploy/helm-charts b/deploy/helm-charts
index dff2e464..3623e156 160000
--- a/deploy/helm-charts
+++ b/deploy/helm-charts
@@ -1,1 +1,1 @@
-Subproject commit dff2e464df614d647c24f77600253ae6efe24e62
+Subproject commit 3623e15654a784f334be59c08ce0fc29a229a0bc
diff --git a/internal/tasks/constants.go b/internal/job/constants.go
similarity index 93%
rename from internal/tasks/constants.go
rename to internal/job/constants.go
index 55cc11dc..ee47349d 100644
--- a/internal/tasks/constants.go
+++ b/internal/job/constants.go
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package tasks
+package job
 
 // Queue Name
 const (
@@ -23,7 +23,7 @@ const (
 	CDNsQueue       = Queue("cdns")
 )
 
-// Task Name
+// Job Name
 const (
-	PreheatTask = "preheat"
+	PreheatJob = "preheat"
 )
diff --git a/internal/tasks/tasks.go b/internal/job/job.go
similarity index 74%
rename from internal/tasks/tasks.go
rename to internal/job/job.go
index c8a447b0..0c17132f 100644
--- a/internal/tasks/tasks.go
+++ b/internal/job/job.go
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package tasks
+package job
 
 import (
 	"encoding/json"
@@ -41,13 +41,13 @@ type Config struct {
 	BackendDB int
 }
 
-type Tasks struct {
+type Job struct {
 	Server *machinery.Server
 	Worker *machinery.Worker
 	Queue  Queue
 }
 
-func New(cfg *Config, queue Queue) (*Tasks, error) {
+func New(cfg *Config, queue Queue) (*Job, error) {
 	broker := fmt.Sprintf("redis://%s@%s:%d/%d", cfg.Password, cfg.Host, cfg.Port, cfg.BrokerDB)
 	backend := fmt.Sprintf("redis://%s@%s:%d/%d", cfg.Password, cfg.Host, cfg.Port, cfg.BackendDB)
 
@@ -63,61 +63,61 @@ func New(cfg *Config, queue Queue) (*Tasks, error) {
 		return nil, err
 	}
 
-	return &Tasks{
+	return &Job{
 		Server: server,
 		Queue:  queue,
 	}, nil
 }
 
-func (t *Tasks) RegisterTasks(namedTaskFuncs map[string]interface{}) error {
-	return t.Server.RegisterTasks(namedTaskFuncs)
+func (t *Job) RegisterJob(namedJobFuncs map[string]interface{}) error {
+	return t.Server.RegisterTasks(namedJobFuncs)
 }
 
-func (t *Tasks) LaunchWorker(consumerTag string, concurrency int) error {
+func (t *Job) LaunchWorker(consumerTag string, concurrency int) error {
 	t.Worker = t.Server.NewWorker(consumerTag, concurrency)
 	return t.Worker.Launch()
 }
 
-type GroupTaskState struct {
+type GroupJobState struct {
 	GroupUUID string
 	State     string
 	CreatedAt time.Time
 }
 
-func (t *Tasks) GetGroupTaskState(groupUUID string) (*GroupTaskState, error) {
-	taskStates, err := t.Server.GetBackend().GroupTaskStates(groupUUID, 0)
+func (t *Job) GetGroupJobState(groupUUID string) (*GroupJobState, error) {
+	jobStates, err := t.Server.GetBackend().GroupTaskStates(groupUUID, 0)
 	if err != nil {
 		return nil, err
 	}
 
-	if len(taskStates) == 0 {
-		return nil, errors.New("empty group task")
+	if len(jobStates) == 0 {
+		return nil, errors.New("empty group job")
 	}
 
-	for _, taskState := range taskStates {
-		if taskState.IsFailure() {
-			return &GroupTaskState{
+	for _, jobState := range jobStates {
+		if jobState.IsFailure() {
+			return &GroupJobState{
 				GroupUUID: groupUUID,
 				State:     machineryv1tasks.StateFailure,
-				CreatedAt: taskState.CreatedAt,
+				CreatedAt: jobState.CreatedAt,
 			}, nil
 		}
 	}
 
-	for _, taskState := range taskStates {
-		if !taskState.IsSuccess() {
-			return &GroupTaskState{
+	for _, jobState := range jobStates {
+		if !jobState.IsSuccess() {
+			return &GroupJobState{
 				GroupUUID: groupUUID,
 				State:     machineryv1tasks.StatePending,
-				CreatedAt: taskState.CreatedAt,
+				CreatedAt: jobState.CreatedAt,
 			}, nil
 		}
 	}
 
-	return &GroupTaskState{
+	return &GroupJobState{
 		GroupUUID: groupUUID,
 		State:     machineryv1tasks.StateSuccess,
-		CreatedAt: taskStates[0].CreatedAt,
+		CreatedAt: jobStates[0].CreatedAt,
 	}, nil
 }
 
diff --git a/internal/tasks/queue.go b/internal/job/queue.go
similarity index 98%
rename from internal/tasks/queue.go
rename to internal/job/queue.go
index 33953a20..db386adf 100644
--- a/internal/tasks/queue.go
+++ b/internal/job/queue.go
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package tasks
+package job
 
 import (
 	"fmt"
diff --git a/internal/tasks/types.go b/internal/job/types.go
similarity index 98%
rename from internal/tasks/types.go
rename to internal/job/types.go
index d0e46ff0..45923f2d 100644
--- a/internal/tasks/types.go
+++ b/internal/job/types.go
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package tasks
+package job
 
 type PreheatRequest struct {
 	URL     string            `json:"url" validate:"required,url"`
diff --git a/manager/tasks/tasks.go b/manager/job/job.go
similarity index 68%
rename from manager/tasks/tasks.go
rename to manager/job/job.go
index 484c67ae..f1951643 100644
--- a/manager/tasks/tasks.go
+++ b/manager/job/job.go
@@ -14,32 +14,32 @@
  * limitations under the License.
  */
 
-package tasks
+package job
 
 import (
-	internaltasks "d7y.io/dragonfly/v2/internal/tasks"
+	internaljob "d7y.io/dragonfly/v2/internal/job"
 	"d7y.io/dragonfly/v2/manager/config"
 	"d7y.io/dragonfly/v2/manager/types"
 )
 
-type Task interface {
+type Job interface {
 	CreatePreheat([]string, types.CreatePreheatRequest) (*types.Preheat, error)
 	GetPreheat(string) (*types.Preheat, error)
 }
 
-type task struct {
-	*internaltasks.Tasks
+type job struct {
+	*internaljob.Job
 	Preheat Preheat
 }
 
-func New(cfg *config.Config) (Task, error) {
-	t, err := internaltasks.New(&internaltasks.Config{
+func New(cfg *config.Config) (Job, error) {
+	t, err := internaljob.New(&internaljob.Config{
 		Host:      cfg.Database.Redis.Host,
 		Port:      cfg.Database.Redis.Port,
 		Password:  cfg.Database.Redis.Password,
 		BrokerDB:  cfg.Database.Redis.BrokerDB,
 		BackendDB: cfg.Database.Redis.BackendDB,
-	}, internaltasks.GlobalQueue)
+	}, internaljob.GlobalQueue)
 	if err != nil {
 		return nil, err
 	}
@@ -49,25 +49,25 @@ func New(cfg *config.Config) (Task, error) {
 		return nil, err
 	}
 
-	return &task{
-		Tasks:   t,
+	return &job{
+		Job:     t,
 		Preheat: p,
 	}, nil
 }
 
-func (t *task) CreatePreheat(hostnames []string, json types.CreatePreheatRequest) (*types.Preheat, error) {
+func (t *job) CreatePreheat(hostnames []string, json types.CreatePreheatRequest) (*types.Preheat, error) {
 	return t.Preheat.CreatePreheat(hostnames, json)
 }
 
-func (t *task) GetPreheat(id string) (*types.Preheat, error) {
-	groupTaskState, err := t.GetGroupTaskState(id)
+func (t *job) GetPreheat(id string) (*types.Preheat, error) {
+	groupJobState, err := t.GetGroupJobState(id)
 	if err != nil {
 		return nil, err
 	}
 
 	return &types.Preheat{
-		ID:        groupTaskState.GroupUUID,
-		Status:    groupTaskState.State,
-		CreatedAt: groupTaskState.CreatedAt,
+		ID:        groupJobState.GroupUUID,
+		Status:    groupJobState.State,
+		CreatedAt: groupJobState.CreatedAt,
 	}, nil
 }
diff --git a/manager/tasks/preheat.go b/manager/job/preheat.go
similarity index 84%
rename from manager/tasks/preheat.go
rename to manager/job/preheat.go
index 9d3c373c..a6fb65fc 100644
--- a/manager/tasks/preheat.go
+++ b/manager/job/preheat.go
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package tasks
+package job
 
 import (
 	"encoding/json"
@@ -26,7 +26,7 @@ import (
 	"strings"
 	"time"
 
-	internaltasks "d7y.io/dragonfly/v2/internal/tasks"
+	internaljob "d7y.io/dragonfly/v2/internal/job"
 	"d7y.io/dragonfly/v2/manager/types"
 	"d7y.io/dragonfly/v2/pkg/util/net/httputils"
 	machineryv1tasks "github.com/RichardKnop/machinery/v1/tasks"
@@ -48,7 +48,7 @@ type Preheat interface {
 }
 
 type preheat struct {
-	tasks  *internaltasks.Tasks
+	job    *internaljob.Job
 	bizTag string
 }
 
@@ -59,9 +59,9 @@ type preheatImage struct {
 	tag      string
 }
 
-func newPreheat(tasks *internaltasks.Tasks, bizTag string) (Preheat, error) {
+func newPreheat(job *internaljob.Job, bizTag string) (Preheat, error) {
 	return &preheat{
-		tasks:  tasks,
+		job:    job,
 		bizTag: bizTag,
 	}, nil
 }
@@ -75,7 +75,7 @@ func (p *preheat) CreatePreheat(hostnames []string, json types.CreatePreheatRequ
 	queues := getSchedulerQueues(hostnames)
 
 	// Generate download files
-	var files []*internaltasks.PreheatRequest
+	var files []*internaljob.PreheatRequest
 	switch PreheatType(json.Type) {
 	case PreheatImageType:
 		// Parse image manifest url
@@ -89,7 +89,7 @@ func (p *preheat) CreatePreheat(hostnames []string, json types.CreatePreheatRequ
 			return nil, err
 		}
 	case PreheatFileType:
-		files = []*internaltasks.PreheatRequest{
+		files = []*internaljob.PreheatRequest{
 			{
 				URL:     url,
 				Tag:     p.bizTag,
@@ -101,20 +101,20 @@ func (p *preheat) CreatePreheat(hostnames []string, json types.CreatePreheatRequ
 		return nil, errors.New("unknow preheat type")
 	}
 
-	return p.createGroupTasks(files, queues)
+	return p.createGroupJob(files, queues)
 }
 
-func (p *preheat) createGroupTasks(files []*internaltasks.PreheatRequest, queues []internaltasks.Queue) (*types.Preheat, error) {
+func (p *preheat) createGroupJob(files []*internaljob.PreheatRequest, queues []internaljob.Queue) (*types.Preheat, error) {
 	signatures := []*machineryv1tasks.Signature{}
 	for _, queue := range queues {
 		for _, file := range files {
-			args, err := internaltasks.MarshalRequest(file)
+			args, err := internaljob.MarshalRequest(file)
 			if err != nil {
 				continue
 			}
 
 			signatures = append(signatures, &machineryv1tasks.Signature{
-				Name:       internaltasks.PreheatTask,
+				Name:       internaljob.PreheatJob,
 				RoutingKey: queue.String(),
 				Args:       args,
 			})
@@ -126,7 +126,7 @@ func (p *preheat) createGroupTasks(files []*internaltasks.PreheatRequest, queues
 		return nil, err
 	}
 
-	if _, err := p.tasks.Server.SendGroup(group, 0); err != nil {
+	if _, err := p.job.Server.SendGroup(group, 0); err != nil {
 		return nil, err
 	}
 
@@ -137,7 +137,7 @@ func (p *preheat) createGroupTasks(files []*internaltasks.PreheatRequest, queues
 	}, nil
 }
 
-func (p *preheat) getLayers(url string, filter string, header http.Header, image *preheatImage) ([]*internaltasks.PreheatRequest, error) {
+func (p *preheat) getLayers(url string, filter string, header http.Header, image *preheatImage) ([]*internaljob.PreheatRequest, error) {
 	resp, err := p.getManifests(url, header)
 	if err != nil {
 		return nil, err
@@ -184,7 +184,7 @@ func (p *preheat) getManifests(url string, header http.Header) (*http.Response,
 	return resp, nil
 }
 
-func (p *preheat) parseLayers(resp *http.Response, filter string, header http.Header, image *preheatImage) ([]*internaltasks.PreheatRequest, error) {
+func (p *preheat) parseLayers(resp *http.Response, filter string, header http.Header, image *preheatImage) ([]*internaljob.PreheatRequest, error) {
 	body, err := ioutil.ReadAll(resp.Body)
 	if err != nil {
 		return nil, err
@@ -195,10 +195,10 @@ func (p *preheat) parseLayers(resp *http.Response, filter string, header http.He
 		return nil, err
 	}
 
-	var layers []*internaltasks.PreheatRequest
+	var layers []*internaljob.PreheatRequest
 	for _, v := range manifest.References() {
 		digest := v.Digest.String()
-		layers = append(layers, &internaltasks.PreheatRequest{
+		layers = append(layers, &internaljob.PreheatRequest{
 			URL:     layerURL(image.protocol, image.domain, image.name, digest),
 			Tag:     p.bizTag,
 			Filter:  filter,
@@ -264,10 +264,10 @@ func parseAccessURL(url string) (*preheatImage, error) {
 	}, nil
 }
 
-func getSchedulerQueues(hostnames []string) []internaltasks.Queue {
-	var queues []internaltasks.Queue
+func getSchedulerQueues(hostnames []string) []internaljob.Queue {
+	var queues []internaljob.Queue
 	for _, hostname := range hostnames {
-		queue, err := internaltasks.GetSchedulerQueue(hostname)
+		queue, err := internaljob.GetSchedulerQueue(hostname)
 		if err != nil {
 			continue
 		}
diff --git a/manager/manager.go b/manager/manager.go
index b47a4b29..c301fff8 100644
--- a/manager/manager.go
+++ b/manager/manager.go
@@ -24,12 +24,12 @@ import (
 	"d7y.io/dragonfly/v2/manager/cache"
 	"d7y.io/dragonfly/v2/manager/config"
 	"d7y.io/dragonfly/v2/manager/database"
+	"d7y.io/dragonfly/v2/manager/job"
 	"d7y.io/dragonfly/v2/manager/permission/rbac"
 	"d7y.io/dragonfly/v2/manager/proxy"
 	"d7y.io/dragonfly/v2/manager/router"
 	"d7y.io/dragonfly/v2/manager/searcher"
 	"d7y.io/dragonfly/v2/manager/service"
-	"d7y.io/dragonfly/v2/manager/tasks"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/manager"
 	"golang.org/x/sync/errgroup"
@@ -67,14 +67,14 @@ func New(cfg *config.Config) (*Server, error) {
 	// Initialize searcher
 	searcher := searcher.New()
 
-	// Initialize task
-	tasks, err := tasks.New(cfg)
+	// Initialize job
+	job, err := job.New(cfg)
 	if err != nil {
 		return nil, err
 	}
 
 	// Initialize REST service
-	restService := service.NewREST(db, cache, tasks, enforcer)
+	restService := service.NewREST(db, cache, job, enforcer)
 
 	// Initialize GRPC service
 	grpcService := service.NewGRPC(db, cache, searcher)
diff --git a/manager/service/preheat.go b/manager/service/preheat.go
index 04a30e1f..1cebfa59 100644
--- a/manager/service/preheat.go
+++ b/manager/service/preheat.go
@@ -36,7 +36,7 @@ func (s *rest) CreatePreheat(json types.CreatePreheatRequest) (*types.Preheat, e
 			return nil, err
 		}
 
-		return s.tasks.CreatePreheat([]string{scheduler.HostName}, json)
+		return s.job.CreatePreheat([]string{scheduler.HostName}, json)
 	}
 
 	schedulerClusters := []model.SchedulerCluster{}
@@ -57,9 +57,9 @@ func (s *rest) CreatePreheat(json types.CreatePreheatRequest) (*types.Preheat, e
 		hostnames = append(hostnames, scheduler.HostName)
 	}
 
-	return s.tasks.CreatePreheat(hostnames, json)
+	return s.job.CreatePreheat(hostnames, json)
 }
 
 func (s *rest) GetPreheat(id string) (*types.Preheat, error) {
-	return s.tasks.GetPreheat(id)
+	return s.job.GetPreheat(id)
 }
diff --git a/manager/service/service.go b/manager/service/service.go
index ba369db3..db40ed3d 100644
--- a/manager/service/service.go
+++ b/manager/service/service.go
@@ -19,8 +19,8 @@ package service
 import (
 	"d7y.io/dragonfly/v2/manager/cache"
 	"d7y.io/dragonfly/v2/manager/database"
+	"d7y.io/dragonfly/v2/manager/job"
 	"d7y.io/dragonfly/v2/manager/model"
-	"d7y.io/dragonfly/v2/manager/tasks"
 	"d7y.io/dragonfly/v2/manager/types"
 	"github.com/casbin/casbin/v2"
 	"github.com/gin-gonic/gin"
@@ -91,17 +91,17 @@ type rest struct {
 	db       *gorm.DB
 	rdb      *redis.Client
 	cache    *cache.Cache
-	tasks    tasks.Task
+	job      job.Job
 	enforcer *casbin.Enforcer
 }
 
 // NewREST returns a new REST instence
-func NewREST(database *database.Database, cache *cache.Cache, tasks tasks.Task, enforcer *casbin.Enforcer) REST {
+func NewREST(database *database.Database, cache *cache.Cache, job job.Job, enforcer *casbin.Enforcer) REST {
 	return &rest{
 		db:       database.DB,
 		rdb:      database.RDB,
 		cache:    cache,
-		tasks:    tasks,
+		job:      job,
 		enforcer: enforcer,
 	}
 }
diff --git a/scheduler/config/config.go b/scheduler/config/config.go
index 42c05c8e..bd5fe513 100644
--- a/scheduler/config/config.go
+++ b/scheduler/config/config.go
@@ -34,7 +34,7 @@ type Config struct {
 	DynConfig    *DynConfig       `yaml:"dynConfig" mapstructure:"dynConfig"`
 	Manager      *ManagerConfig   `yaml:"manager" mapstructure:"manager"`
 	Host         *HostConfig      `yaml:"host" mapstructure:"host"`
-	Task         *TaskConfig      `yaml:"task" mapstructure:"task"`
+	Job          *JobConfig       `yaml:"job" mapstructure:"job"`
 }
 
 func New() *Config {
@@ -44,7 +44,7 @@ func New() *Config {
 		DynConfig: NewDefaultDynConfig(),
 		Manager:   NewDefaultManagerConfig(),
 		Host:      NewHostConfig(),
-		Task:      NewDefaultTaskConfig(),
+		Job:       NewDefaultJobConfig(),
 	}
 }
 
@@ -151,8 +151,8 @@ func NewDefaultManagerConfig() *ManagerConfig {
 	}
 }
 
-func NewDefaultTaskConfig() *TaskConfig {
-	return &TaskConfig{
+func NewDefaultJobConfig() *JobConfig {
+	return &JobConfig{
 		GlobalWorkerNum:    10,
 		SchedulerWorkerNum: 10,
 		LocalWorkerNum:     10,
@@ -167,12 +167,12 @@ func NewDefaultTaskConfig() *TaskConfig {
 }
 
 func (c *Config) Convert() error {
-	if c.Manager.Addr != "" && c.Task.Redis.Host == "" {
+	if c.Manager.Addr != "" && c.Job.Redis.Host == "" {
 		host, _, err := net.SplitHostPort(c.Manager.Addr)
 		if err != nil {
 			return err
 		}
-		c.Task.Redis.Host = host
+		c.Job.Redis.Host = host
 	}
 	return nil
 }
@@ -264,7 +264,7 @@ type RedisConfig struct {
 	BackendDB int    `yaml:"backendDB" mapstructure:"backendDB"`
 }
 
-type TaskConfig struct {
+type JobConfig struct {
 	GlobalWorkerNum    uint         `yaml:"globalWorkerNum" mapstructure:"globalWorkerNum"`
 	SchedulerWorkerNum uint         `yaml:"schedulerWorkerNum" mapstructure:"schedulerWorkerNum"`
 	LocalWorkerNum     uint         `yaml:"localWorkerNum" mapstructure:"localWorkerNum"`
diff --git a/scheduler/tasks/tasks.go b/scheduler/job/job.go
similarity index 59%
rename from scheduler/tasks/tasks.go
rename to scheduler/job/job.go
index f5e29c67..01888ec0 100644
--- a/scheduler/tasks/tasks.go
+++ b/scheduler/job/job.go
@@ -1,4 +1,4 @@
-package tasks
+package job
 
 import (
 	"context"
@@ -10,7 +10,7 @@ import (
 	"d7y.io/dragonfly/v2/internal/dferrors"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/internal/idgen"
-	internaltasks "d7y.io/dragonfly/v2/internal/tasks"
+	internaljob "d7y.io/dragonfly/v2/internal/job"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/scheduler/config"
 	"d7y.io/dragonfly/v2/scheduler/core"
@@ -23,22 +23,22 @@ const (
 	interval = time.Second
 )
 
-type Tasks interface {
+type Job interface {
 	Serve() error
 	Stop()
 }
 
-type tasks struct {
-	globalTasks    *internaltasks.Tasks
-	schedulerTasks *internaltasks.Tasks
-	localTasks     *internaltasks.Tasks
-	ctx            context.Context
-	service        *core.SchedulerService
-	cfg            *config.TaskConfig
+type job struct {
+	globalJob    *internaljob.Job
+	schedulerJob *internaljob.Job
+	localJob     *internaljob.Job
+	ctx          context.Context
+	service      *core.SchedulerService
+	cfg          *config.JobConfig
 }
 
-func New(ctx context.Context, cfg *config.TaskConfig, hostname string, service *core.SchedulerService) (Tasks, error) {
-	redisConfig := &internaltasks.Config{
+func New(ctx context.Context, cfg *config.JobConfig, hostname string, service *core.SchedulerService) (Job, error) {
+	redisConfig := &internaljob.Config{
 		Host:      cfg.Redis.Host,
 		Port:      cfg.Redis.Port,
 		Password:  cfg.Redis.Password,
@@ -46,56 +46,56 @@ func New(ctx context.Context, cfg *config.TaskConfig, hostname string, service *
 		BackendDB: cfg.Redis.BackendDB,
 	}
 
-	globalTask, err := internaltasks.New(redisConfig, internaltasks.GlobalQueue)
+	globalJob, err := internaljob.New(redisConfig, internaljob.GlobalQueue)
 	if err != nil {
-		logger.Errorf("create global tasks queue error: %v", err)
+		logger.Errorf("create global job queue error: %v", err)
 		return nil, err
 	}
 
-	schedulerTask, err := internaltasks.New(redisConfig, internaltasks.SchedulersQueue)
+	schedulerJob, err := internaljob.New(redisConfig, internaljob.SchedulersQueue)
 	if err != nil {
-		logger.Errorf("create scheduler tasks queue error: %v", err)
+		logger.Errorf("create scheduler job queue error: %v", err)
 		return nil, err
 	}
 
-	localQueue, err := internaltasks.GetSchedulerQueue(hostname)
+	localQueue, err := internaljob.GetSchedulerQueue(hostname)
 	if err != nil {
-		logger.Errorf("get local tasks queue name error: %v", err)
+		logger.Errorf("get local job queue name error: %v", err)
 		return nil, err
 	}
 
-	localTask, err := internaltasks.New(redisConfig, localQueue)
+	localJob, err := internaljob.New(redisConfig, localQueue)
 	if err != nil {
-		logger.Errorf("create local tasks queue error: %v", err)
+		logger.Errorf("create local job queue error: %v", err)
 		return nil, err
 	}
 
-	t := &tasks{
-		globalTasks:    globalTask,
-		schedulerTasks: schedulerTask,
-		localTasks:     localTask,
-		ctx:            ctx,
-		service:        service,
-		cfg:            cfg,
+	t := &job{
+		globalJob:    globalJob,
+		schedulerJob: schedulerJob,
+		localJob:     localJob,
+		ctx:          ctx,
+		service:      service,
+		cfg:          cfg,
 	}
 
-	namedTaskFuncs := map[string]interface{}{
-		internaltasks.PreheatTask: t.preheat,
+	namedJobFuncs := map[string]interface{}{
+		internaljob.PreheatJob: t.preheat,
 	}
 
-	if err := localTask.RegisterTasks(namedTaskFuncs); err != nil {
-		logger.Errorf("register preheat tasks to local queue error: %v", err)
+	if err := localJob.RegisterJob(namedJobFuncs); err != nil {
+		logger.Errorf("register preheat job to local queue error: %v", err)
 		return nil, err
 	}
 
 	return t, nil
 }
 
-func (t *tasks) Serve() error {
+func (t *job) Serve() error {
 	g := errgroup.Group{}
 	g.Go(func() error {
 		logger.Debugf("ready to launch %d worker(s) on global queue", t.cfg.GlobalWorkerNum)
-		err := t.globalTasks.LaunchWorker("global_worker", int(t.cfg.GlobalWorkerNum))
+		err := t.globalJob.LaunchWorker("global_worker", int(t.cfg.GlobalWorkerNum))
 		if err != nil {
 			logger.Errorf("global queue worker error: %v", err)
 		}
@@ -104,7 +104,7 @@ func (t *tasks) Serve() error {
 
 	g.Go(func() error {
 		logger.Debugf("ready to launch %d worker(s) on scheduler queue", t.cfg.SchedulerWorkerNum)
-		err := t.schedulerTasks.LaunchWorker("scheduler_worker", int(t.cfg.SchedulerWorkerNum))
+		err := t.schedulerJob.LaunchWorker("scheduler_worker", int(t.cfg.SchedulerWorkerNum))
 		if err != nil {
 			logger.Errorf("scheduler queue worker error: %v", err)
 		}
@@ -113,7 +113,7 @@ func (t *tasks) Serve() error {
 
 	g.Go(func() error {
 		logger.Debugf("ready to launch %d worker(s) on local queue", t.cfg.LocalWorkerNum)
-		err := t.localTasks.LaunchWorker("local_worker", int(t.cfg.LocalWorkerNum))
+		err := t.localJob.LaunchWorker("local_worker", int(t.cfg.LocalWorkerNum))
 		if err != nil {
 			logger.Errorf("local queue worker error: %v", err)
 		}
@@ -125,15 +125,15 @@ func (t *tasks) Serve() error {
 	return werr
 }
 
-func (t *tasks) Stop() {
-	t.globalTasks.Worker.Quit()
-	t.schedulerTasks.Worker.Quit()
-	t.localTasks.Worker.Quit()
+func (t *job) Stop() {
+	t.globalJob.Worker.Quit()
+	t.schedulerJob.Worker.Quit()
+	t.localJob.Worker.Quit()
 }
 
-func (t *tasks) preheat(req string) error {
-	request := &internaltasks.PreheatRequest{}
-	if err := internaltasks.UnmarshalRequest(req, request); err != nil {
+func (t *job) preheat(req string) error {
+	request := &internaljob.PreheatRequest{}
+	if err := internaljob.UnmarshalRequest(req, request); err != nil {
 		logger.Errorf("unmarshal request err: %v, request body: %s", err, req)
 		return err
 	}
diff --git a/scheduler/scheduler.go b/scheduler/scheduler.go
index 62bd257f..3d27cfc3 100644
--- a/scheduler/scheduler.go
+++ b/scheduler/scheduler.go
@@ -20,8 +20,8 @@ import (
 	"context"
 	"time"
 
+	"d7y.io/dragonfly/v2/scheduler/job"
 	server2 "d7y.io/dragonfly/v2/scheduler/server"
-	"d7y.io/dragonfly/v2/scheduler/tasks"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 
 	"d7y.io/dragonfly/v2/cmd/dependency"
@@ -49,7 +49,7 @@ type Server struct {
 	dynconfigConn    *grpc.ClientConn
 	running          bool
 	dynConfig        config.DynconfigInterface
-	task             tasks.Tasks
+	job              job.Job
 }
 
 func New(cfg *config.Config) (*Server, error) {
@@ -106,8 +106,8 @@ func New(cfg *config.Config) (*Server, error) {
 	if err != nil {
 		return nil, err
 	}
-	if cfg.Task.Redis.Host != "" {
-		s.task, err = tasks.New(context.Background(), cfg.Task, iputils.HostName, s.schedulerService)
+	if cfg.Job.Redis.Host != "" {
+		s.job, err = job.New(context.Background(), cfg.Job, iputils.HostName, s.schedulerService)
 		if err != nil {
 			return nil, err
 		}
@@ -125,8 +125,8 @@ func (s *Server) Serve() error {
 	s.dynConfig.Serve()
 	s.schedulerService.Serve()
 
-	if s.task != nil {
-		go s.task.Serve()
+	if s.job != nil {
+		go s.job.Serve()
 	}
 
 	if s.managerClient != nil {
@@ -163,7 +163,7 @@ func (s *Server) Stop() (err error) {
 		s.dynConfig.Stop()
 		rpc.StopServer()
 		s.schedulerService.Stop()
-		s.task.Stop()
+		s.job.Stop()
 	}
 	return
 }


Test patch:
diff --git a/client/daemon/peer/peertask_manager_mock_test.go b/client/daemon/peer/peertask_manager_mock_test.go
index fd0a0fb2..d4ea5432 100644
--- a/client/daemon/peer/peertask_manager_mock_test.go
+++ b/client/daemon/peer/peertask_manager_mock_test.go
@@ -1,19 +1,3 @@
-/*
- *     Copyright 2020 The Dragonfly Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
 // Code generated by MockGen. DO NOT EDIT.
 // Source: peertask_manager.go
 
@@ -24,294 +8,359 @@ import (
 	context "context"
 	io "io"
 	reflect "reflect"
-
-	gomock "github.com/golang/mock/gomock"
+	time "time"
 
 	dflog "d7y.io/dragonfly/v2/internal/dflog"
 	base "d7y.io/dragonfly/v2/pkg/rpc/base"
 	scheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler"
+	gomock "github.com/golang/mock/gomock"
 )
 
-// MockPeerTaskManager is a mock of PeerTaskManager interface
-type MockPeerTaskManager struct {
+// MockTaskManager is a mock of TaskManager interface.
+type MockTaskManager struct {
 	ctrl     *gomock.Controller
-	recorder *MockPeerTaskManagerMockRecorder
+	recorder *MockTaskManagerMockRecorder
 }
 
-// MockPeerTaskManagerMockRecorder is the mock recorder for MockPeerTaskManager
-type MockPeerTaskManagerMockRecorder struct {
-	mock *MockPeerTaskManager
+// MockTaskManagerMockRecorder is the mock recorder for MockTaskManager.
+type MockTaskManagerMockRecorder struct {
+	mock *MockTaskManager
 }
 
-// NewMockPeerTaskManager creates a new mock instance
-func NewMockPeerTaskManager(ctrl *gomock.Controller) *MockPeerTaskManager {
-	mock := &MockPeerTaskManager{ctrl: ctrl}
-	mock.recorder = &MockPeerTaskManagerMockRecorder{mock}
+// NewMockTaskManager creates a new mock instance.
+func NewMockTaskManager(ctrl *gomock.Controller) *MockTaskManager {
+	mock := &MockTaskManager{ctrl: ctrl}
+	mock.recorder = &MockTaskManagerMockRecorder{mock}
 	return mock
 }
 
-// EXPECT returns an object that allows the caller to indicate expected use
-func (m *MockPeerTaskManager) EXPECT() *MockPeerTaskManagerMockRecorder {
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockTaskManager) EXPECT() *MockTaskManagerMockRecorder {
 	return m.recorder
 }
 
-// StartFilePeerTask mocks base method
-func (m *MockPeerTaskManager) StartFilePeerTask(ctx context.Context, req *FilePeerTaskRequest) (chan *FilePeerTaskProgress, error) {
+// IsPeerTaskRunning mocks base method.
+func (m *MockTaskManager) IsPeerTaskRunning(pid string) bool {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "IsPeerTaskRunning", pid)
+	ret0, _ := ret[0].(bool)
+	return ret0
+}
+
+// IsPeerTaskRunning indicates an expected call of IsPeerTaskRunning.
+func (mr *MockTaskManagerMockRecorder) IsPeerTaskRunning(pid interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsPeerTaskRunning", reflect.TypeOf((*MockTaskManager)(nil).IsPeerTaskRunning), pid)
+}
+
+// StartFilePeerTask mocks base method.
+func (m *MockTaskManager) StartFilePeerTask(ctx context.Context, req *FilePeerTaskRequest) (chan *FilePeerTaskProgress, *TinyData, error) {
+	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "StartFilePeerTask", ctx, req)
 	ret0, _ := ret[0].(chan *FilePeerTaskProgress)
-	ret1, _ := ret[1].(error)
-	return ret0, ret1
+	ret1, _ := ret[1].(*TinyData)
+	ret2, _ := ret[2].(error)
+	return ret0, ret1, ret2
 }
 
-// StartFilePeerTask indicates an expected call of StartFilePeerTask
-func (mr *MockPeerTaskManagerMockRecorder) StartFilePeerTask(ctx, req interface{}) *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "StartFilePeerTask", reflect.TypeOf((*MockPeerTaskManager)(nil).StartFilePeerTask), ctx, req)
+// StartFilePeerTask indicates an expected call of StartFilePeerTask.
+func (mr *MockTaskManagerMockRecorder) StartFilePeerTask(ctx, req interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "StartFilePeerTask", reflect.TypeOf((*MockTaskManager)(nil).StartFilePeerTask), ctx, req)
 }
 
-// StartStreamPeerTask mocks base method
-func (m *MockPeerTaskManager) StartStreamPeerTask(ctx context.Context, req *scheduler.PeerTaskRequest) (io.Reader, map[string]string, error) {
+// StartStreamPeerTask mocks base method.
+func (m *MockTaskManager) StartStreamPeerTask(ctx context.Context, req *scheduler.PeerTaskRequest) (io.ReadCloser, map[string]string, error) {
+	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "StartStreamPeerTask", ctx, req)
-	ret0, _ := ret[0].(io.Reader)
+	ret0, _ := ret[0].(io.ReadCloser)
 	ret1, _ := ret[1].(map[string]string)
 	ret2, _ := ret[2].(error)
 	return ret0, ret1, ret2
 }
 
-// StartStreamPeerTask indicates an expected call of StartStreamPeerTask
-func (mr *MockPeerTaskManagerMockRecorder) StartStreamPeerTask(ctx, req interface{}) *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "StartStreamPeerTask", reflect.TypeOf((*MockPeerTaskManager)(nil).StartStreamPeerTask), ctx, req)
+// StartStreamPeerTask indicates an expected call of StartStreamPeerTask.
+func (mr *MockTaskManagerMockRecorder) StartStreamPeerTask(ctx, req interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "StartStreamPeerTask", reflect.TypeOf((*MockTaskManager)(nil).StartStreamPeerTask), ctx, req)
 }
 
-// Stop mocks base method
-func (m *MockPeerTaskManager) Stop(ctx context.Context) error {
+// Stop mocks base method.
+func (m *MockTaskManager) Stop(ctx context.Context) error {
+	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "Stop", ctx)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
-// Stop indicates an expected call of Stop
-func (mr *MockPeerTaskManagerMockRecorder) Stop(ctx interface{}) *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Stop", reflect.TypeOf((*MockPeerTaskManager)(nil).Stop), ctx)
+// Stop indicates an expected call of Stop.
+func (mr *MockTaskManagerMockRecorder) Stop(ctx interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Stop", reflect.TypeOf((*MockTaskManager)(nil).Stop), ctx)
 }
 
-// MockPeerTask is a mock of PeerTask interface
-type MockPeerTask struct {
+// MockTask is a mock of Task interface.
+type MockTask struct {
 	ctrl     *gomock.Controller
-	recorder *MockPeerTaskMockRecorder
+	recorder *MockTaskMockRecorder
 }
 
-// MockPeerTaskMockRecorder is the mock recorder for MockPeerTask
-type MockPeerTaskMockRecorder struct {
-	mock *MockPeerTask
+// MockTaskMockRecorder is the mock recorder for MockTask.
+type MockTaskMockRecorder struct {
+	mock *MockTask
 }
 
-// NewMockPeerTask creates a new mock instance
-func NewMockPeerTask(ctrl *gomock.Controller) *MockPeerTask {
-	mock := &MockPeerTask{ctrl: ctrl}
-	mock.recorder = &MockPeerTaskMockRecorder{mock}
+// NewMockTask creates a new mock instance.
+func NewMockTask(ctrl *gomock.Controller) *MockTask {
+	mock := &MockTask{ctrl: ctrl}
+	mock.recorder = &MockTaskMockRecorder{mock}
 	return mock
 }
 
-// EXPECT returns an object that allows the caller to indicate expected use
-func (m *MockPeerTask) EXPECT() *MockPeerTaskMockRecorder {
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockTask) EXPECT() *MockTaskMockRecorder {
 	return m.recorder
 }
 
-// ReportPieceResult mocks base method
-func (m *MockPeerTask) ReportPieceResult(pieceTask *base.PieceInfo, pieceResult *scheduler.PieceResult) error {
-	ret := m.ctrl.Call(m, "ReportPieceResult", pieceTask, pieceResult)
-	ret0, _ := ret[0].(error)
-	return ret0
+// AddTraffic mocks base method.
+func (m *MockTask) AddTraffic(arg0 int64) {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "AddTraffic", arg0)
 }
 
-// ReportPieceResult indicates an expected call of ReportPieceResult
-func (mr *MockPeerTaskMockRecorder) ReportPieceResult(pieceTask, pieceResult interface{}) *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReportPieceResult", reflect.TypeOf((*MockPeerTask)(nil).ReportPieceResult), pieceTask, pieceResult)
+// AddTraffic indicates an expected call of AddTraffic.
+func (mr *MockTaskMockRecorder) AddTraffic(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddTraffic", reflect.TypeOf((*MockTask)(nil).AddTraffic), arg0)
 }
 
-// GetPeerID mocks base method
-func (m *MockPeerTask) GetPeerID() string {
-	ret := m.ctrl.Call(m, "GetPeerID")
-	ret0, _ := ret[0].(string)
+// Context mocks base method.
+func (m *MockTask) Context() context.Context {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Context")
+	ret0, _ := ret[0].(context.Context)
 	return ret0
 }
 
-// GetPeerID indicates an expected call of GetPeerID
-func (mr *MockPeerTaskMockRecorder) GetPeerID() *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPeerID", reflect.TypeOf((*MockPeerTask)(nil).GetPeerID))
+// Context indicates an expected call of Context.
+func (mr *MockTaskMockRecorder) Context() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Context", reflect.TypeOf((*MockTask)(nil).Context))
 }
 
-// GetTaskID mocks base method
-func (m *MockPeerTask) GetTaskID() string {
-	ret := m.ctrl.Call(m, "GetTaskID")
-	ret0, _ := ret[0].(string)
+// GetContentLength mocks base method.
+func (m *MockTask) GetContentLength() int64 {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetContentLength")
+	ret0, _ := ret[0].(int64)
 	return ret0
 }
 
-// GetTaskID indicates an expected call of GetTaskID
-func (mr *MockPeerTaskMockRecorder) GetTaskID() *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTaskID", reflect.TypeOf((*MockPeerTask)(nil).GetTaskID))
+// GetContentLength indicates an expected call of GetContentLength.
+func (mr *MockTaskMockRecorder) GetContentLength() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetContentLength", reflect.TypeOf((*MockTask)(nil).GetContentLength))
 }
 
-// GetTotalPieces mocks base method
-func (m *MockPeerTask) GetTotalPieces() int32 {
-	ret := m.ctrl.Call(m, "GetTotalPieces")
-	ret0, _ := ret[0].(int32)
+// GetPeerID mocks base method.
+func (m *MockTask) GetPeerID() string {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetPeerID")
+	ret0, _ := ret[0].(string)
 	return ret0
 }
 
-// GetTotalPieces indicates an expected call of GetTotalPieces
-func (mr *MockPeerTaskMockRecorder) GetTotalPieces() *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTotalPieces", reflect.TypeOf((*MockPeerTask)(nil).GetTotalPieces))
+// GetPeerID indicates an expected call of GetPeerID.
+func (mr *MockTaskMockRecorder) GetPeerID() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPeerID", reflect.TypeOf((*MockTask)(nil).GetPeerID))
 }
 
-// GetContentLength mocks base method
-func (m *MockPeerTask) GetContentLength() int64 {
-	ret := m.ctrl.Call(m, "GetContentLength")
-	ret0, _ := ret[0].(int64)
+// GetTaskID mocks base method.
+func (m *MockTask) GetTaskID() string {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetTaskID")
+	ret0, _ := ret[0].(string)
 	return ret0
 }
 
-// GetContentLength indicates an expected call of GetContentLength
-func (mr *MockPeerTaskMockRecorder) GetContentLength() *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetContentLength", reflect.TypeOf((*MockPeerTask)(nil).GetContentLength))
+// GetTaskID indicates an expected call of GetTaskID.
+func (mr *MockTaskMockRecorder) GetTaskID() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTaskID", reflect.TypeOf((*MockTask)(nil).GetTaskID))
 }
 
-// SetContentLength mocks base method
-func (m *MockPeerTask) SetContentLength(arg0 int64) error {
-	ret := m.ctrl.Call(m, "SetContentLength", arg0)
-	ret0, _ := ret[0].(error)
+// GetTotalPieces mocks base method.
+func (m *MockTask) GetTotalPieces() int32 {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetTotalPieces")
+	ret0, _ := ret[0].(int32)
 	return ret0
 }
 
-// SetContentLength indicates an expected call of SetContentLength
-func (mr *MockPeerTaskMockRecorder) SetContentLength(arg0 interface{}) *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetContentLength", reflect.TypeOf((*MockPeerTask)(nil).SetContentLength), arg0)
+// GetTotalPieces indicates an expected call of GetTotalPieces.
+func (mr *MockTaskMockRecorder) GetTotalPieces() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTotalPieces", reflect.TypeOf((*MockTask)(nil).GetTotalPieces))
 }
 
-// SetCallback mocks base method
-func (m *MockPeerTask) SetCallback(arg0 TaskCallback) {
-	m.ctrl.Call(m, "SetCallback", arg0)
+// GetTraffic mocks base method.
+func (m *MockTask) GetTraffic() int64 {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetTraffic")
+	ret0, _ := ret[0].(int64)
+	return ret0
 }
 
-// SetCallback indicates an expected call of SetCallback
-func (mr *MockPeerTaskMockRecorder) SetCallback(arg0 interface{}) *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetCallback", reflect.TypeOf((*MockPeerTask)(nil).SetCallback), arg0)
+// GetTraffic indicates an expected call of GetTraffic.
+func (mr *MockTaskMockRecorder) GetTraffic() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTraffic", reflect.TypeOf((*MockTask)(nil).GetTraffic))
 }
 
-// AddTraffic mocks base method
-func (m *MockPeerTask) AddTraffic(arg0 int64) {
-	m.ctrl.Call(m, "AddTraffic", arg0)
+// Log mocks base method.
+func (m *MockTask) Log() *dflog.SugaredLoggerOnWith {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Log")
+	ret0, _ := ret[0].(*dflog.SugaredLoggerOnWith)
+	return ret0
 }
 
-// AddTraffic indicates an expected call of AddTraffic
-func (mr *MockPeerTaskMockRecorder) AddTraffic(arg0 interface{}) *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddTraffic", reflect.TypeOf((*MockPeerTask)(nil).AddTraffic), arg0)
+// Log indicates an expected call of Log.
+func (mr *MockTaskMockRecorder) Log() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Log", reflect.TypeOf((*MockTask)(nil).Log))
 }
 
-// GetTraffic mocks base method
-func (m *MockPeerTask) GetTraffic() int64 {
-	ret := m.ctrl.Call(m, "GetTraffic")
-	ret0, _ := ret[0].(int64)
+// ReportPieceResult mocks base method.
+func (m *MockTask) ReportPieceResult(result *pieceTaskResult) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ReportPieceResult", result)
+	ret0, _ := ret[0].(error)
 	return ret0
 }
 
-// GetTraffic indicates an expected call of GetTraffic
-func (mr *MockPeerTaskMockRecorder) GetTraffic() *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTraffic", reflect.TypeOf((*MockPeerTask)(nil).GetTraffic))
+// ReportPieceResult indicates an expected call of ReportPieceResult.
+func (mr *MockTaskMockRecorder) ReportPieceResult(result interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReportPieceResult", reflect.TypeOf((*MockTask)(nil).ReportPieceResult), result)
 }
 
-// GetContext mocks base method
-func (m *MockPeerTask) Context() context.Context {
-	ret := m.ctrl.Call(m, "Context")
-	ret0, _ := ret[0].(context.Context)
-	return ret0
+// SetCallback mocks base method.
+func (m *MockTask) SetCallback(arg0 TaskCallback) {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "SetCallback", arg0)
 }
 
-// GetContext indicates an expected call of GetContext
-func (mr *MockPeerTaskMockRecorder) GetContext() *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Context", reflect.TypeOf((*MockPeerTask)(nil).Context))
+// SetCallback indicates an expected call of SetCallback.
+func (mr *MockTaskMockRecorder) SetCallback(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetCallback", reflect.TypeOf((*MockTask)(nil).SetCallback), arg0)
 }
 
-// Log mocks base method
-func (m *MockPeerTask) Log() *dflog.SugaredLoggerOnWith {
-	ret := m.ctrl.Call(m, "Log")
-	ret0, _ := ret[0].(*dflog.SugaredLoggerOnWith)
+// SetContentLength mocks base method.
+func (m *MockTask) SetContentLength(arg0 int64) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "SetContentLength", arg0)
+	ret0, _ := ret[0].(error)
 	return ret0
 }
 
-// Log indicates an expected call of Log
-func (mr *MockPeerTaskMockRecorder) Log() *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Log", reflect.TypeOf((*MockPeerTask)(nil).Log))
+// SetContentLength indicates an expected call of SetContentLength.
+func (mr *MockTaskMockRecorder) SetContentLength(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetContentLength", reflect.TypeOf((*MockTask)(nil).SetContentLength), arg0)
 }
 
-// MockPeerTaskCallback is a mock of PeerTaskCallback interface
-type MockPeerTaskCallback struct {
+// MockTaskCallback is a mock of TaskCallback interface.
+type MockTaskCallback struct {
 	ctrl     *gomock.Controller
-	recorder *MockPeerTaskCallbackMockRecorder
+	recorder *MockTaskCallbackMockRecorder
 }
 
-// MockPeerTaskCallbackMockRecorder is the mock recorder for MockPeerTaskCallback
-type MockPeerTaskCallbackMockRecorder struct {
-	mock *MockPeerTaskCallback
+// MockTaskCallbackMockRecorder is the mock recorder for MockTaskCallback.
+type MockTaskCallbackMockRecorder struct {
+	mock *MockTaskCallback
 }
 
-// NewMockPeerTaskCallback creates a new mock instance
-func NewMockPeerTaskCallback(ctrl *gomock.Controller) *MockPeerTaskCallback {
-	mock := &MockPeerTaskCallback{ctrl: ctrl}
-	mock.recorder = &MockPeerTaskCallbackMockRecorder{mock}
+// NewMockTaskCallback creates a new mock instance.
+func NewMockTaskCallback(ctrl *gomock.Controller) *MockTaskCallback {
+	mock := &MockTaskCallback{ctrl: ctrl}
+	mock.recorder = &MockTaskCallbackMockRecorder{mock}
 	return mock
 }
 
-// EXPECT returns an object that allows the caller to indicate expected use
-func (m *MockPeerTaskCallback) EXPECT() *MockPeerTaskCallbackMockRecorder {
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockTaskCallback) EXPECT() *MockTaskCallbackMockRecorder {
 	return m.recorder
 }
 
-// Init mocks base method
-func (m *MockPeerTaskCallback) Init(pt Task) error {
-	ret := m.ctrl.Call(m, "Init", pt)
+// Done mocks base method.
+func (m *MockTaskCallback) Done(pt Task) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Done", pt)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
-// Init indicates an expected call of Init
-func (mr *MockPeerTaskCallbackMockRecorder) Init(pt interface{}) *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Init", reflect.TypeOf((*MockPeerTaskCallback)(nil).Init), pt)
+// Done indicates an expected call of Done.
+func (mr *MockTaskCallbackMockRecorder) Done(pt interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Done", reflect.TypeOf((*MockTaskCallback)(nil).Done), pt)
 }
 
-// Done mocks base method
-func (m *MockPeerTaskCallback) Done(pt Task) error {
-	ret := m.ctrl.Call(m, "Done", pt)
+// Fail mocks base method.
+func (m *MockTaskCallback) Fail(pt Task, code base.Code, reason string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Fail", pt, code, reason)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
-// Done indicates an expected call of Done
-func (mr *MockPeerTaskCallbackMockRecorder) Done(pt interface{}) *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Done", reflect.TypeOf((*MockPeerTaskCallback)(nil).Done), pt)
+// Fail indicates an expected call of Fail.
+func (mr *MockTaskCallbackMockRecorder) Fail(pt, code, reason interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Fail", reflect.TypeOf((*MockTaskCallback)(nil).Fail), pt, code, reason)
 }
 
-// Update mocks base method
-func (m *MockPeerTaskCallback) Update(pt Task) error {
-	ret := m.ctrl.Call(m, "Update", pt)
+// GetStartTime mocks base method.
+func (m *MockTaskCallback) GetStartTime() time.Time {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetStartTime")
+	ret0, _ := ret[0].(time.Time)
+	return ret0
+}
+
+// GetStartTime indicates an expected call of GetStartTime.
+func (mr *MockTaskCallbackMockRecorder) GetStartTime() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetStartTime", reflect.TypeOf((*MockTaskCallback)(nil).GetStartTime))
+}
+
+// Init mocks base method.
+func (m *MockTaskCallback) Init(pt Task) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Init", pt)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
-// Update indicates an expected call of Update
-func (mr *MockPeerTaskCallbackMockRecorder) Update(pt interface{}) *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Update", reflect.TypeOf((*MockPeerTaskCallback)(nil).Update), pt)
+// Init indicates an expected call of Init.
+func (mr *MockTaskCallbackMockRecorder) Init(pt interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Init", reflect.TypeOf((*MockTaskCallback)(nil).Init), pt)
 }
 
-// Fail mocks base method
-func (m *MockPeerTaskCallback) Fail(pt Task, reason string) error {
-	ret := m.ctrl.Call(m, "Fail", pt, reason)
+// Update mocks base method.
+func (m *MockTaskCallback) Update(pt Task) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Update", pt)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
-// Fail indicates an expected call of Fail
-func (mr *MockPeerTaskCallbackMockRecorder) Fail(pt, reason interface{}) *gomock.Call {
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Fail", reflect.TypeOf((*MockPeerTaskCallback)(nil).Fail), pt, reason)
+// Update indicates an expected call of Update.
+func (mr *MockTaskCallbackMockRecorder) Update(pt interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Update", reflect.TypeOf((*MockTaskCallback)(nil).Update), pt)
 }
diff --git a/client/daemon/peer/piece_manager_test.go b/client/daemon/peer/piece_manager_test.go
index 462ea18d..e5fe8e44 100644
--- a/client/daemon/peer/piece_manager_test.go
+++ b/client/daemon/peer/piece_manager_test.go
@@ -123,9 +123,9 @@ func TestPieceManager_DownloadSource(t *testing.T) {
 		},
 	}
 	for _, tc := range testCases {
-		func() {
+		t.Run(tc.name, func(t *testing.T) {
 			/********** prepare test start **********/
-			mockPeerTask := NewMockPeerTask(ctrl)
+			mockPeerTask := NewMockTask(ctrl)
 			mockPeerTask.EXPECT().SetContentLength(gomock.Any()).AnyTimes().DoAndReturn(
 				func(arg0 int64) error {
 					return nil
@@ -139,11 +139,11 @@ func TestPieceManager_DownloadSource(t *testing.T) {
 					return taskID
 				})
 			mockPeerTask.EXPECT().AddTraffic(gomock.Any()).AnyTimes().DoAndReturn(func(int642 int64) {})
-			mockPeerTask.EXPECT().ReportPieceResult(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(
-				func(pieceTask *base.PieceInfo, pieceResult *scheduler.PieceResult) error {
+			mockPeerTask.EXPECT().ReportPieceResult(gomock.Any()).AnyTimes().DoAndReturn(
+				func(result *pieceTaskResult) error {
 					return nil
 				})
-			mockPeerTask.EXPECT().GetContext().AnyTimes().DoAndReturn(func() context.Context {
+			mockPeerTask.EXPECT().Context().AnyTimes().DoAndReturn(func() context.Context {
 				return context.Background()
 			})
 			mockPeerTask.EXPECT().Log().AnyTimes().DoAndReturn(func() *logger.SugaredLoggerOnWith {
@@ -162,8 +162,6 @@ func TestPieceManager_DownloadSource(t *testing.T) {
 			defer storageManager.CleanUp()
 			defer os.Remove(output)
 			/********** prepare test end **********/
-
-			t.Logf("test case: %s", tc.name)
 			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 				if tc.withContentLength {
 					w.Header().Set("Content-Length",
@@ -209,6 +207,6 @@ func TestPieceManager_DownloadSource(t *testing.T) {
 			outputBytes, err := ioutil.ReadFile(output)
 			assert.Nil(err, "load output file")
 			assert.Equal(testBytes, outputBytes, "output and desired output must match")
-		}()
+		})
 	}
 }
diff --git a/internal/tasks/tasks_test.go b/internal/job/job_test.go
similarity index 98%
rename from internal/tasks/tasks_test.go
rename to internal/job/job_test.go
index 69f03cf8..12042684 100644
--- a/internal/tasks/tasks_test.go
+++ b/internal/job/job_test.go
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package tasks
+package job
 
 import (
 	"reflect"
@@ -25,7 +25,7 @@ import (
 	machineryv1tasks "github.com/RichardKnop/machinery/v1/tasks"
 )
 
-func TestTaskMarshal(t *testing.T) {
+func TestJobMarshal(t *testing.T) {
 	tests := []struct {
 		name   string
 		value  interface{}
@@ -111,7 +111,7 @@ func TestTaskMarshal(t *testing.T) {
 	}
 }
 
-func TestTaskUnmarshal(t *testing.T) {
+func TestJobUnmarshal(t *testing.T) {
 	tests := []struct {
 		name   string
 		data   []reflect.Value
diff --git a/internal/tasks/queue_test.go b/internal/job/queue_test.go
similarity index 95%
rename from internal/tasks/queue_test.go
rename to internal/job/queue_test.go
index 6f8bc065..ec2ea3ae 100644
--- a/internal/tasks/queue_test.go
+++ b/internal/job/queue_test.go
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package tasks
+package job
 
 import (
 	"testing"
@@ -22,7 +22,7 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestTaskGetSchedulerQueue(t *testing.T) {
+func TestJobGetSchedulerQueue(t *testing.T) {
 	tests := []struct {
 		name     string
 		hostname string
@@ -54,7 +54,7 @@ func TestTaskGetSchedulerQueue(t *testing.T) {
 	}
 }
 
-func TestTaskGetCDNQueue(t *testing.T) {
+func TestJobGetCDNQueue(t *testing.T) {
 	tests := []struct {
 		name     string
 		hostname string
diff --git a/scheduler/config/config_test.go b/scheduler/config/config_test.go
index cca8baef..6113f73c 100644
--- a/scheduler/config/config_test.go
+++ b/scheduler/config/config_test.go
@@ -64,7 +64,7 @@ func TestSchedulerConfig_Load(t *testing.T) {
 			IDC:      "foo",
 			Location: "bar",
 		},
-		Task: &TaskConfig{
+		Job: &JobConfig{
 			GlobalWorkerNum:    1,
 			SchedulerWorkerNum: 1,
 			LocalWorkerNum:     5,
@@ -98,7 +98,7 @@ func TestConvert(t *testing.T) {
 				Manager: &ManagerConfig{
 					Addr: "127.0.0.1:65003",
 				},
-				Task: &TaskConfig{
+				Job: &JobConfig{
 					Redis: &RedisConfig{
 						Host: "",
 					},
@@ -106,7 +106,7 @@ func TestConvert(t *testing.T) {
 			},
 			expect: func(t *testing.T, cfg *Config, err error) {
 				assert := assert.New(t)
-				assert.Equal("127.0.0.1", cfg.Task.Redis.Host)
+				assert.Equal("127.0.0.1", cfg.Job.Redis.Host)
 			},
 		},
 		{
@@ -115,7 +115,7 @@ func TestConvert(t *testing.T) {
 				Manager: &ManagerConfig{
 					Addr: "127.0.0.1:65003",
 				},
-				Task: &TaskConfig{
+				Job: &JobConfig{
 					Redis: &RedisConfig{
 						Host: "111.111.11.1",
 					},
@@ -123,7 +123,7 @@ func TestConvert(t *testing.T) {
 			},
 			expect: func(t *testing.T, cfg *Config, err error) {
 				assert := assert.New(t)
-				assert.Equal("111.111.11.1", cfg.Task.Redis.Host)
+				assert.Equal("111.111.11.1", cfg.Job.Redis.Host)
 			},
 		},
 		{
@@ -132,7 +132,7 @@ func TestConvert(t *testing.T) {
 				Manager: &ManagerConfig{
 					Addr: "",
 				},
-				Task: &TaskConfig{
+				Job: &JobConfig{
 					Redis: &RedisConfig{
 						Host: "111.111.11.1",
 					},
@@ -140,7 +140,7 @@ func TestConvert(t *testing.T) {
 			},
 			expect: func(t *testing.T, cfg *Config, err error) {
 				assert := assert.New(t)
-				assert.Equal("111.111.11.1", cfg.Task.Redis.Host)
+				assert.Equal("111.111.11.1", cfg.Job.Redis.Host)
 			},
 		},
 		{
@@ -149,7 +149,7 @@ func TestConvert(t *testing.T) {
 				Manager: &ManagerConfig{
 					Addr: ":65003",
 				},
-				Task: &TaskConfig{
+				Job: &JobConfig{
 					Redis: &RedisConfig{
 						Host: "",
 					},
@@ -157,7 +157,7 @@ func TestConvert(t *testing.T) {
 			},
 			expect: func(t *testing.T, cfg *Config, err error) {
 				assert := assert.New(t)
-				assert.Equal("", cfg.Task.Redis.Host)
+				assert.Equal("", cfg.Job.Redis.Host)
 			},
 		},
 		{
@@ -166,7 +166,7 @@ func TestConvert(t *testing.T) {
 				Manager: &ManagerConfig{
 					Addr: "localhost:65003",
 				},
-				Task: &TaskConfig{
+				Job: &JobConfig{
 					Redis: &RedisConfig{
 						Host: "",
 					},
@@ -174,7 +174,7 @@ func TestConvert(t *testing.T) {
 			},
 			expect: func(t *testing.T, cfg *Config, err error) {
 				assert := assert.New(t)
-				assert.Equal("localhost", cfg.Task.Redis.Host)
+				assert.Equal("localhost", cfg.Job.Redis.Host)
 			},
 		},
 	}
diff --git a/scheduler/config/testdata/scheduler.yaml b/scheduler/config/testdata/scheduler.yaml
index 04fed18b..8d1c0bdb 100644
--- a/scheduler/config/testdata/scheduler.yaml
+++ b/scheduler/config/testdata/scheduler.yaml
@@ -41,7 +41,7 @@ host:
   idc: foo
   location: bar
 
-task:
+job:
   globalWorkerNum: 1
   schedulerWorkerNum: 1
   localWorkerNum: 5