func (s *Service) RegisterPeerTask(ctx context.Context, req *rpcscheduler.PeerTaskRequest) (*rpcscheduler.RegisterResult, error) {
	// Register task and trigger cdn download task
	task, err := s.registerTask(ctx, req)
	if err != nil {
		dferr := dferrors.New(base.Code_SchedTaskStatusError, "register task is fail")
		logger.Errorf("peer %s register is failed: %v", req.PeerId, err)
		return nil, dferr
	}
	host := s.registerHost(ctx, req)
	peer := s.registerPeer(ctx, req, task, host)
	peer.Log.Infof("register peer task request: %#v", req)

	// Task has been successful
	if task.FSM.Is(resource.TaskStateSucceeded) {
		peer.Log.Info("task has been successful")
		sizeScope := task.SizeScope()
		switch sizeScope {
		case base.SizeScope_TINY:
			peer.Log.Info("task size scope is tiny and return piece content directly")
			if len(task.DirectPiece) > 0 && int64(len(task.DirectPiece)) == task.ContentLength.Load() {
				if err := peer.FSM.Event(resource.PeerEventRegisterTiny); err != nil {
					dferr := dferrors.New(base.Code_SchedError, err.Error())
					peer.Log.Errorf("peer %s register is failed: %v", req.PeerId, err)
					return nil, dferr
				}

				return &rpcscheduler.RegisterResult{
					TaskId:    task.ID,
					SizeScope: base.SizeScope_TINY,
					DirectPiece: &rpcscheduler.RegisterResult_PieceContent{
						PieceContent: task.DirectPiece,
					},
				}, nil
			}

			// Fallback to base.SizeScope_SMALL
			peer.Log.Warnf("task size scope is tiny, length of direct piece is %d and content length is %d. fall through to size scope small",
				len(task.DirectPiece), task.ContentLength.Load())
			fallthrough
		case base.SizeScope_SMALL:
			peer.Log.Info("task size scope is small")
			// If the file is registered as a small type,
			// there is no need to build a tree, just find the parent and return
			parent, ok := s.scheduler.FindParent(ctx, peer, set.NewSafeSet())
			if !ok {
				peer.Log.Warn("task size scope is small and it can not select parent")
				if err := peer.FSM.Event(resource.PeerEventRegisterNormal); err != nil {
					dferr := dferrors.New(base.Code_SchedError, err.Error())
					peer.Log.Errorf("peer %s register is failed: %v", req.PeerId, err)
					return nil, dferr
				}

				return &rpcscheduler.RegisterResult{
					TaskId:    task.ID,
					SizeScope: base.SizeScope_NORMAL,
				}, nil
			}

			firstPiece, ok := task.LoadPiece(0)
			if !ok {
				peer.Log.Warn("task size scope is small and it can not get first piece")
				if err := peer.FSM.Event(resource.PeerEventRegisterNormal); err != nil {
					dferr := dferrors.New(base.Code_SchedError, err.Error())
					peer.Log.Errorf("peer %s register is failed: %v", req.PeerId, err)
					return nil, dferr
				}

				return &rpcscheduler.RegisterResult{
					TaskId:    task.ID,
					SizeScope: base.SizeScope_NORMAL,
				}, nil
			}

			peer.ReplaceParent(parent)
			if err := peer.FSM.Event(resource.PeerEventRegisterSmall); err != nil {
				dferr := dferrors.New(base.Code_SchedError, err.Error())
				peer.Log.Errorf("peer %s register is failed: %v", req.PeerId, err)
				return nil, dferr
			}

			singlePiece := &rpcscheduler.SinglePiece{
				DstPid:  parent.ID,
				DstAddr: fmt.Sprintf("%s:%d", parent.Host.IP, parent.Host.DownloadPort),
				PieceInfo: &base.PieceInfo{
					PieceNum:    firstPiece.PieceNum,
					RangeStart:  firstPiece.RangeStart,
					RangeSize:   firstPiece.RangeSize,
					PieceMd5:    firstPiece.PieceMd5,
					PieceOffset: firstPiece.PieceOffset,
					PieceStyle:  firstPiece.PieceStyle,
				},
			}

			peer.Log.Infof("task size scope is small and return single piece: %#v %#v", singlePiece, singlePiece.PieceInfo)
			return &rpcscheduler.RegisterResult{
				TaskId:    task.ID,
				SizeScope: base.SizeScope_SMALL,
				DirectPiece: &rpcscheduler.RegisterResult_SinglePiece{
					SinglePiece: singlePiece,
				},
			}, nil
		default:
			peer.Log.Info("task size scope is normal and needs to be register")
			if err := peer.FSM.Event(resource.PeerEventRegisterNormal); err != nil {
				dferr := dferrors.New(base.Code_SchedError, err.Error())
				peer.Log.Errorf("peer %s register is failed: %v", req.PeerId, err)
				return nil, dferr
			}

			return &rpcscheduler.RegisterResult{
				TaskId:    task.ID,
				SizeScope: base.SizeScope_NORMAL,
			}, nil
		}
	}

	// Task is unsuccessful
	peer.Log.Info("task is unsuccessful and needs to be register")
	if err := peer.FSM.Event(resource.PeerEventRegisterNormal); err != nil {
		dferr := dferrors.New(base.Code_SchedError, err.Error())
		peer.Log.Errorf("peer %s register is failed: %v", req.PeerId, err)
		return nil, dferr
	}

	return &rpcscheduler.RegisterResult{
		TaskId:    task.ID,
		SizeScope: base.SizeScope_NORMAL,
	}, nil
}
// RegisterPeerTask registers peer and triggers CDN download task
RegisterPeerTask registers peer and triggers CDN download task
// RegisterPeerTask registers peer and triggers CDN download task
func (s *Service) RegisterPeerTask(ctx context.Context, req *rpcscheduler.PeerTaskRequest) (*rpcscheduler.RegisterResult, error) {
	task, err := s.registerTask(ctx, req)
	if err != nil {
		dferr := dferrors.New(base.Code_SchedTaskStatusError, "register task is fail")
		logger.Errorf("peer %s register is failed: %v", req.PeerId, err)
		return nil, dferr
	}
	host := s.registerHost(ctx, req)
	peer := s.registerPeer(ctx, req, task, host)
	peer.Log.Infof("register peer task request: %#v", req)
	if task.FSM.Is(resource.TaskStateSucceeded) {
		peer.Log.Info("task has been successful")
		sizeScope := task.SizeScope()
		switch sizeScope {
		case base.SizeScope_TINY:
			peer.Log.Info("task size scope is tiny and return piece content directly")
			if len(task.DirectPiece) > 0 && int64(len(task.DirectPiece)) == task.ContentLength.Load() {
				if err := peer.FSM.Event(resource.PeerEventRegisterTiny); err != nil {
					dferr := dferrors.New(base.Code_SchedError, err.Error())
					peer.Log.Errorf("peer %s register is failed: %v", req.PeerId, err)
					return nil, dferr
				}
				return &rpcscheduler.RegisterResult{
					TaskId:    task.ID,
					SizeScope: base.SizeScope_TINY,
					DirectPiece: &rpcscheduler.RegisterResult_PieceContent{
						PieceContent: task.DirectPiece,
					},
				}, nil
			}
			peer.Log.Warnf("task size scope is tiny, length of direct piece is %d and content length is %d. fall through to size scope small",
				len(task.DirectPiece), task.ContentLength.Load())
			fallthrough
		case base.SizeScope_SMALL:
			peer.Log.Info("task size scope is small")
			parent, ok := s.scheduler.FindParent(ctx, peer, set.NewSafeSet())
			if !ok {
				peer.Log.Warn("task size scope is small and it can not select parent")
				if err := peer.FSM.Event(resource.PeerEventRegisterNormal); err != nil {
					dferr := dferrors.New(base.Code_SchedError, err.Error())
					peer.Log.Errorf("peer %s register is failed: %v", req.PeerId, err)
					return nil, dferr
				}
				return &rpcscheduler.RegisterResult{
					TaskId:    task.ID,
					SizeScope: base.SizeScope_NORMAL,
				}, nil
			}
			firstPiece, ok := task.LoadPiece(0)
			if !ok {
				peer.Log.Warn("task size scope is small and it can not get first piece")
				if err := peer.FSM.Event(resource.PeerEventRegisterNormal); err != nil {
					dferr := dferrors.New(base.Code_SchedError, err.Error())
					peer.Log.Errorf("peer %s register is failed: %v", req.PeerId, err)
					return nil, dferr
				}
				return &rpcscheduler.RegisterResult{
					TaskId:    task.ID,
					SizeScope: base.SizeScope_NORMAL,
				}, nil
			}
			peer.ReplaceParent(parent)
			if err := peer.FSM.Event(resource.PeerEventRegisterSmall); err != nil {
				dferr := dferrors.New(base.Code_SchedError, err.Error())
				peer.Log.Errorf("peer %s register is failed: %v", req.PeerId, err)
				return nil, dferr
			}
			singlePiece := &rpcscheduler.SinglePiece{
				DstPid:  parent.ID,
				DstAddr: fmt.Sprintf("%s:%d", parent.Host.IP, parent.Host.DownloadPort),
				PieceInfo: &base.PieceInfo{
					PieceNum:    firstPiece.PieceNum,
					RangeStart:  firstPiece.RangeStart,
					RangeSize:   firstPiece.RangeSize,
					PieceMd5:    firstPiece.PieceMd5,
					PieceOffset: firstPiece.PieceOffset,
					PieceStyle:  firstPiece.PieceStyle,
				},
			}
			peer.Log.Infof("task size scope is small and return single piece: %#v %#v", singlePiece, singlePiece.PieceInfo)
			return &rpcscheduler.RegisterResult{
				TaskId:    task.ID,
				SizeScope: base.SizeScope_SMALL,
				DirectPiece: &rpcscheduler.RegisterResult_SinglePiece{
					SinglePiece: singlePiece,
				},
			}, nil
		default:
			peer.Log.Info("task size scope is normal and needs to be register")
			if err := peer.FSM.Event(resource.PeerEventRegisterNormal); err != nil {
				dferr := dferrors.New(base.Code_SchedError, err.Error())
				peer.Log.Errorf("peer %s register is failed: %v", req.PeerId, err)
				return nil, dferr
			}
			return &rpcscheduler.RegisterResult{
				TaskId:    task.ID,
				SizeScope: base.SizeScope_NORMAL,
			}, nil
		}
	}
	peer.Log.Info("task is unsuccessful and needs to be register")
	if err := peer.FSM.Event(resource.PeerEventRegisterNormal); err != nil {
		dferr := dferrors.New(base.Code_SchedError, err.Error())
		peer.Log.Errorf("peer %s register is failed: %v", req.PeerId, err)
		return nil, dferr
	}
	return &rpcscheduler.RegisterResult{
		TaskId:    task.ID,
		SizeScope: base.SizeScope_NORMAL,
	}, nil
}
func ( s * Service ) RegisterPeerTask ( ctx context . Context , req * rpcscheduler . PeerTaskRequest ) ( * rpcscheduler . RegisterResult , error ) { task , err := s . registerTask ( ctx , req ) if err != nil { dferr := dferrors . New ( base . Code_SchedTaskStatusError , " " ) logger . Errorf ( " " , req . PeerId , err ) return nil , dferr } host := s . registerHost ( ctx , req ) peer := s . registerPeer ( ctx , req , task , host ) peer . Log . Infof ( " " , req ) if task . FSM . Is ( resource . TaskStateSucceeded ) { peer . Log . Info ( " " ) sizeScope := task . SizeScope ( ) switch sizeScope { case base . SizeScope_TINY : peer . Log . Info ( " " ) if len ( task . DirectPiece ) > 0 && int64 ( len ( task . DirectPiece ) ) == task . ContentLength . Load ( ) { if err := peer . FSM . Event ( resource . PeerEventRegisterTiny ) ; err != nil { dferr := dferrors . New ( base . Code_SchedError , err . Error ( ) ) peer . Log . Errorf ( " " , req . PeerId , err ) return nil , dferr } return & rpcscheduler . RegisterResult { TaskId : task . ID , SizeScope : base . SizeScope_TINY , DirectPiece : & rpcscheduler . RegisterResult_PieceContent { PieceContent : task . DirectPiece , } , } , nil } peer . Log . Warnf ( " " , len ( task . DirectPiece ) , task . ContentLength . Load ( ) ) fallthrough case base . SizeScope_SMALL : peer . Log . Info ( " " ) parent , ok := s . scheduler . FindParent ( ctx , peer , set . NewSafeSet ( ) ) if ! ok { peer . Log . Warn ( " " ) if err := peer . FSM . Event ( resource . PeerEventRegisterNormal ) ; err != nil { dferr := dferrors . New ( base . Code_SchedError , err . Error ( ) ) peer . Log . Errorf ( " " , req . PeerId , err ) return nil , dferr } return & rpcscheduler . RegisterResult { TaskId : task . ID , SizeScope : base . SizeScope_NORMAL , } , nil } firstPiece , ok := task . LoadPiece ( 0 ) if ! ok { peer . Log . Warn ( " " ) if err := peer . FSM . Event ( resource . PeerEventRegisterNormal ) ; err != nil { dferr := dferrors . New ( base . Code_SchedError , err . Error ( ) ) peer . Log . Errorf ( " " , req . PeerId , err ) return nil , dferr } return & rpcscheduler . RegisterResult { TaskId : task . ID , SizeScope : base . SizeScope_NORMAL , } , nil } peer . ReplaceParent ( parent ) if err := peer . FSM . Event ( resource . PeerEventRegisterSmall ) ; err != nil { dferr := dferrors . New ( base . Code_SchedError , err . Error ( ) ) peer . Log . Errorf ( " " , req . PeerId , err ) return nil , dferr } singlePiece := & rpcscheduler . SinglePiece { DstPid : parent . ID , DstAddr : fmt . Sprintf ( " " , parent . Host . IP , parent . Host . DownloadPort ) , PieceInfo : & base . PieceInfo { PieceNum : firstPiece . PieceNum , RangeStart : firstPiece . RangeStart , RangeSize : firstPiece . RangeSize , PieceMd5 : firstPiece . PieceMd5 , PieceOffset : firstPiece . PieceOffset , PieceStyle : firstPiece . PieceStyle , } , } peer . Log . Infof ( " " , singlePiece , singlePiece . PieceInfo ) return & rpcscheduler . RegisterResult { TaskId : task . ID , SizeScope : base . SizeScope_SMALL , DirectPiece : & rpcscheduler . RegisterResult_SinglePiece { SinglePiece : singlePiece , } , } , nil default : peer . Log . Info ( " " ) if err := peer . FSM . Event ( resource . PeerEventRegisterNormal ) ; err != nil { dferr := dferrors . New ( base . Code_SchedError , err . Error ( ) ) peer . Log . Errorf ( " " , req . PeerId , err ) return nil , dferr } return & rpcscheduler . RegisterResult { TaskId : task . ID , SizeScope : base . SizeScope_NORMAL , } , nil } } peer . Log . Info ( " " ) if err := peer . FSM . Event ( resource . PeerEventRegisterNormal ) ; err != nil { dferr := dferrors . New ( base . Code_SchedError , err . Error ( ) ) peer . Log . Errorf ( " " , req . PeerId , err ) return nil , dferr } return & rpcscheduler . RegisterResult { TaskId : task . ID , SizeScope : base . SizeScope_NORMAL , } , nil }
RegisterPeerTask registers peer and triggers CDN download task
RegisterPeerTask registers peer and triggers CDN download task
// Register task and trigger cdn download task // Task has been successful // Fallback to base.SizeScope_SMALL // If the file is registered as a small type, // there is no need to build a tree, just find the parent and return // Task is unsuccessful